#include "mytcpsocket.h"
#include <QMetaObject>
#include <memory>
#include <QtEndian>

extern QUEUE_DATA<MESG> queue_send;
extern QUEUE_DATA<MESG> queue_recv;
extern QUEUE_DATA<MESG> audio_recv;

MyTcpSocket::MyTcpSocket(QObject *parent) : QThread(parent){
    qRegisterMetaType<QAbstractSocket::SocketError>();
    _socktcp = nullptr;

    _sockThread = new QThread();
    this->moveToThread(_sockThread);
    connect(_sockThread, &QThread::finished, this, &MyTcpSocket::closeSocket);

    sendbuf = (uchar *)malloc(4 * MB);
    recvbuf = (uchar *)malloc(4 * MB);
    hasrecvive = 0;
}

void MyTcpSocket::closeSocket(){
    if(_socktcp && _socktcp->isOpen()){
        _socktcp->close();
    }
}

void MyTcpSocket::errorDetect(QAbstractSocket::SocketError error){
    qDebug() << "Sock error" << QThread::currentThreadId();
    std::shared_ptr<MESG> msg = std::make_shared<MESG>();
    memset(msg.get(), 0 ,sizeof(MESG));

    if(error == QAbstractSocket::RemoteHostClosedError){
        msg->msg_type = RemoteHostClosedError;
    }else{
        msg->msg_type = OtherNetError;
    }

    queue_recv.push_msg(msg.get());
}

void MyTcpSocket::sendData(MESG *send){
    if(_socktcp->state() == QAbstractSocket::UnconnectedState){
        emit sendTextOver();
        if(send->data) free(send->data);
        if(send) free(send);
        return ;
    }
    //消息头
    quint64 bytetowrite = 0;
    sendbuf[bytetowrite++] = '$';

    //消息类型
    qToBigEndian<quint16>(send->msg_type, sendbuf + bytetowrite);
    bytetowrite += 2;

    //发送者IP
    quint32 ip = _socktcp->localAddress().toIPv4Address();
    qToBigEndian<quint32>(ip, sendbuf + bytetowrite);
    bytetowrite += 4;

    //创建会议，发送音频，关闭摄像头，发送图片，
    if(send->msg_type == CREATE_MEETING || send->msg_type == AUDIO_SEND || send->msg_type == CLOSE_CAMERA || send->msg_type == IMG_SEND || send->msg_type == TEXT_SEND){
        qToBigEndian<quint32>(send->len, sendbuf + bytetowrite);
        bytetowrite += 4;
    }else if(send->msg_type == JOIN_MEETING){
        qToBigEndian<quint32>(send->len, sendbuf + bytetowrite);
        bytetowrite += 4;
        uint32_t room;
        memcpy(&room, send->data, send->len);
        qToBigEndian<quint32>(room, send->data);
    }

    //将数据拷入sendbuf
    memcpy(sendbuf + bytetowrite, send->data, send->len);
    bytetowrite += send->len;
    sendbuf[bytetowrite++] = '#';

    qint64 hastowrite = bytetowrite;
    qint64 ret = 0, haswrite = 0;
    int retryCount = 0;
    const int MAX_RETRIES = 10;

    while(haswrite < hastowrite){
        ret = _socktcp->write((char *)sendbuf + haswrite, hastowrite - haswrite);

        if(ret > 0){
            haswrite += ret;
        }else if(ret == 0){
            qDebug() << "fwrite returned 0 without error";
            break;
        }else{
            if(errno == EAGAIN || errno == EWOULDBLOCK){
                if (retryCount++ >= MAX_RETRIES) {
                    qDebug() << "Reached maximum retries for EAGAIN/EWOULDBLOCK";
                    break;
                }
                QThread::msleep(1);
                ret = 0;
            }else{
                qDebug() << "fwrite error:" << strerror(errno);
            }
        }
    }
    if (haswrite < hastowrite) {
        qDebug() << "Failed to write entire log. Wrote" << haswrite << "bytes out of" << hastowrite;
    }

    _socktcp->waitForBytesWritten();

    if(send->msg_type == TEXT_SEND){
        emit sendTextOver(); //成功往内核发送文本信息
    }

    if (send->data) free(send->data);
    if (send) free(send);
}

void MyTcpSocket::run(){
    m_isCanRun = true;
    for(;;){
        {
            QMutexLocker locker(&m_lock);
            if(m_isCanRun == false) return ;
        }

        MESG *send = queue_send.pop_msg();
        if(send == nullptr) continue;
        QMetaObject::invokeMethod(this, "sendData", Q_ARG(MESG*, send));
    }
}

qint64 MyTcpSocket::readn(char *buf, quint64 maxsize, int n){
    quint64 hastoread = n;
    quint64 hasread = 0;
    do{
        qint64 ret = _socktcp->read(buf + hasread, hastoread);
        if(ret < 0){
            return -1;
        }else if(ret == 0){
            return hasread;
        }
        hastoread -= ret;
        hasread += ret;
    }while(hastoread > 0 && hasread < maxsize);
    return hasread;
}

void MyTcpSocket::recvFromSocket(){
    qint64 availbytes = _socktcp->bytesAvailable();
    if(availbytes <= 0){
        return ;
    }
    qint64 ret = _socktcp->read((char *)recvbuf + hasrecvive, availbytes);
    if(ret <= 0){
        qDebug() << "error or no more data";
        return ;
    }
    hasrecvive += ret;

    //数据包不够
    if(hasrecvive < MSG_HEADER){
        return ;
    }else{
        quint32 data_size;
        qFromBigEndian<quint32>(recvbuf + 7, 4, &data_size);
        if((quint64)data_size + 1 + MSG_HEADER <= hasrecvive){
            if(recvbuf[0] == '$' && recvbuf[MSG_HEADER + data_size] == '#'){
                MSG_TYPE msgtype;
                uint16_t type;
                qFromBigEndian<uint16_t>(recvbuf + 1, 2, &type);
                msgtype = (MSG_TYPE)type;
                qDebug() << "recv data type: " << msgtype;
                if(msgtype == CREATE_MEETING_RESPONSE || msgtype == JOIN_MEETING_RESPONSE || msgtype == PARTNER_JOIN2){
                    if(msgtype == CREATE_MEETING_RESPONSE){
                        qint32 roomNo;
                        qFromBigEndian<qint32>(recvbuf + MSG_HEADER, 4, &roomNo);

                        MESG *msg = (MESG *)malloc(sizeof(MESG));
                        if(msg == nullptr){
                            qDebug() << __LINE__ << " CREATE_MEETING_RESPONSE malloc MESG failed";
                        }else{
                            memset(msg, 0, sizeof(MESG));
                            msg->msg_type = msgtype;
                            msg->data = (uchar*)malloc((quint64)data_size);
                            if(msg->data == nullptr){
                                free(msg);
                                qDebug() << __LINE__ << "CREATE_MEETING_RESPONSE malloc MESG.data failed";
                            }else{
                                memset(msg->data, 0, (quint64)data_size);
                                memcpy(msg->data, &roomNo, data_size);
                                msg->len = data_size;
                                queue_recv.push_msg(msg);
                            }
                        }
                    }else if(msgtype == JOIN_MEETING_RESPONSE){
                        qint32 c;
                        memcpy(&c, recvbuf + MSG_HEADER, data_size);
                        MESG *msg = (MESG *)malloc(sizeof(MESG));
                        if(msg == nullptr){
                            return ;
                        }else{
                            memset(msg, 0, sizeof(MESG));
                            msg->msg_type = msgtype;
                            msg->data = (uchar*)malloc(data_size);
                            if(msg->data == nullptr){
                                free(msg);
                                qDebug() << __LINE__ << "JOIN_MEETING_RESPONSE malloc MESG.data failed";
                            }else{
                                memset(msg->data, 0, data_size);
                                memcpy(msg->data, &c, data_size);
                                msg->len = data_size;
                                queue_recv.push_msg(msg);
                            }
                        }
                    }else if(msgtype == PARTNER_JOIN2){
                        MESG *msg = (MESG*)malloc(sizeof(MESG));
                        if(msg == nullptr){
                            qDebug() << "PARTNER_JOIN2 malloc MESG error";
                        }else{
                            memset(msg, 0, sizeof(MESG));
                            msg->msg_type = msgtype;
                            msg->len = data_size;
                            msg->data = (uchar*)malloc(data_size);
                            if(msg->data == nullptr){
                                free(msg);
                                qDebug() << "PARTNER_JOIN2 malloc MESG.data error";
                            }else{
                                memset(msg->data, 0, data_size);
                                uint32_t ip;
                                int pos = 0;
                                for(int i = 0; i < data_size / sizeof(uint32_t); ++i){
                                    qFromBigEndian<uint32_t>(recvbuf + MSG_HEADER + pos, sizeof(uint32_t), &ip);
                                    memcpy_s(msg->data + pos, data_size - pos, &ip, sizeof(uint32_t));
                                    pos += sizeof(uint32_t);
                                }
                                queue_recv.push_msg(msg);
                            }
                        }
                    }
                }else if(msgtype == IMG_RECV || msgtype == PARTNER_JOIN || msgtype == PARTNER_EXIT || msgtype == AUDIO_RECV || msgtype == CLOSE_CAMERA || msgtype == TEXT_RECV){
                    quint32 ip;
                    qFromBigEndian<quint32>(recvbuf + 3, 4, &ip);

                    if(msgtype == IMG_RECV){
                        QByteArray cc((char *)recvbuf + MSG_HEADER, data_size);
                        QByteArray rc = QByteArray::fromBase64(cc);
                        QByteArray rdc = qUncompress(rc);

                        if(rdc.size() > 0){
                            MESG *msg = (MESG*)malloc(sizeof(MESG));
                            if(msg == nullptr){
                                qDebug() << __LINE__ << " malloc failed";
                            }else{
                                memset(msg, 0, sizeof(MESG));
                                msg->msg_type = msgtype;
                                msg->data = (uchar*)malloc(rdc.size());
                                if(msg->data == nullptr){
                                    free(msg);
                                    qDebug() << __LINE__ << " malloc failed";
                                }else{
                                    memset(msg->data, 0, rdc.size());
                                    memcpy_s(msg->data, rdc.size(), rdc.data(), rdc.size());
                                    msg->len = rdc.size();
                                    msg->ip = ip;
                                    queue_recv.push_msg(msg);
                                }
                            }
                        }
                    }else if(msgtype == PARTNER_JOIN || msgtype == PARTNER_EXIT || msgtype ==CLOSE_CAMERA){
                        MESG *msg = (MESG*)malloc(sizeof(MESG));
                        if(msg == nullptr){
                            qDebug() << __LINE__ << " malloc failed";
                        }else{
                            memset(msg, 0, sizeof(MESG));
                            msg->msg_type = msgtype;
                            msg->ip = ip;
                            queue_recv.push_msg(msg);
                        }
                    }else if(msgtype == AUDIO_RECV){
                        QByteArray cc((char*)recvbuf + MSG_HEADER, data_size);
                        QByteArray rc = QByteArray::fromBase64(cc);
                        QByteArray rdc = qUncompress(rc);

                        if(rdc.size() > 0){
                            MESG *msg = (MESG*)malloc(sizeof(MESG));
                            if(msg == nullptr){
                                qDebug() << __LINE__ << "malloc failed";
                            }else{
                                memset(msg, 0, sizeof(MESG));
                                msg->msg_type = AUDIO_RECV;
                                msg->ip = ip;

                                msg->data = (uchar*)malloc(rdc.size());
                                if(msg->data == nullptr){
                                    free(msg);
                                    qDebug() << __LINE__ << "malloc msg.data failed";
                                }else{
                                    memset(msg->data, 0, rdc.size());
                                    memcpy_s(msg->data, rdc.size(), rdc.data(), rdc.size());
                                    msg->len = rdc.size();
                                    msg->ip = ip;
                                    audio_recv.push_msg(msg);
                                }
                            }
                        }
                    }else if(msgtype == TEXT_RECV){
                        QByteArray cc((char*)recvbuf + MSG_HEADER, data_size);
                        std::string rr = qUncompress(cc).toStdString();
                        if(rr.size() > 0){
                            MESG *msg = (MESG *)malloc(sizeof(MESG));
                            if(msg == nullptr){
                                qDebug() << __LINE__ << "malloc failed";
                            }else{
                                memset(msg, 0, sizeof(MESG));
                                msg->msg_type = TEXT_RECV;
                                msg->ip = ip;
                                msg->data = (uchar*)malloc(rr.size());
                                if(msg->data == nullptr){
                                    free(msg);
                                    qDebug() << __LINE__ << "malloc msg.data failed";
                                }else{
                                    memset(msg->data, 0, rr.size());
                                    memcpy_s(msg->data, rr.size(), rr.data(), rr.size());
                                    msg->len = rr.size();
                                    queue_recv.push_msg(msg);
                                }
                            }
                        }
                    }
                }
            }else{
                 qDebug() << "package error";
            }
            memmove_s(recvbuf, 4 * MB, recvbuf + MSG_HEADER + data_size + 1, hasrecvive - ((quint64)data_size + 1 + MSG_HEADER));
            hasrecvive -= ((quint64)data_size + 1 + MSG_HEADER);
        }else{
            return ;
        }
    }
}

MyTcpSocket::~MyTcpSocket(){
    delete sendbuf;
    delete _sockThread;
}

bool MyTcpSocket::connectServer(QString ip, QString port, QIODevice::OpenModeFlag flag){
    if(_socktcp == nullptr) _socktcp = new QTcpSocket();
    _socktcp->connectToHost(ip, port.toUShort(), flag);
    connect(_socktcp, &QTcpSocket::readyRead, this, &MyTcpSocket::recvFromSocket, Qt::UniqueConnection);
    connect(_socktcp, &QTcpSocket::errorOccurred, this, &MyTcpSocket::errorDetect, Qt::UniqueConnection);

    if(_socktcp->waitForConnected(5000)){
        return true;
    }
    _socktcp->close();
    return false;
}

bool MyTcpSocket::connectToServer(QString ip, QString port, QIODevice::OpenModeFlag flag){
    _sockThread->start();
    bool retVal;
    QMetaObject::invokeMethod(this, "connectServer", Qt::BlockingQueuedConnection, Q_RETURN_ARG(bool, retVal),\
                              Q_ARG(QString, ip), Q_ARG(QString, port), Q_ARG(QIODevice::OpenModeFlag, flag));
    if(retVal){
        this->start();
        return true;
    }  else{
        return false;
    }
}

QString MyTcpSocket::errorString(){
    return _socktcp->errorString();
}

void MyTcpSocket::disconnectFromHost(){
    if(this->isRunning()){
        QMutexLocker<QMutex> locker(&m_lock);
        m_isCanRun = false;
    }
    if(_sockThread->isRunning()){
        _sockThread->quit();
        _sockThread->wait();
    }
    queue_send.clear();
    queue_recv.clear();
    audio_recv.clear();
}

quint32 MyTcpSocket::getlocalip(){
    if(_socktcp->isOpen()){
        return _socktcp->localAddress().toIPv4Address();
    }else{
        return -1;
    }
}

void MyTcpSocket::stopImmediately(){
    {
        QMutexLocker<QMutex> lock(&m_lock);
        if(m_isCanRun == true) m_isCanRun = false;

        _sockThread->quit();
        _sockThread->wait();
    }
}
